package com.atguigu.gmall.realtime.app

import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.utils.{MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * 业务数据的分流
 * 1.准备实时环境
 * 2.从redis中读取offset
 * 3.从Kafka中消费数据
 * 4.提取偏移量
 * 5.处理数据
 *      5.1 转换结构
 *      5.2 分流
 *          维度数据 ->redis
 *          事实数据 ->kafka
 *              以表为准，以操作为单位进行分流
 *              例如：order_info 表新增数据  DWD_ORDER_INFO_I 主题
 *              例如：order_info 表修改数据  DWD_ORDER_INFO_U 主题
 *              例如：order_info 表删除数据  DWD_ORDER_INFO_D
 * 6.flush缓冲区
 * 7.提交offset到redis中
 *
 */
object OdsBaseDbApp {
  def main(args: Array[String]): Unit = {
    //1.准备streaming环境
    val conf: SparkConf = new SparkConf().setAppName("ods_Base_Db_Log").setMaster("local[4]")
    val ssc = new StreamingContext(conf,Seconds(5))

    //2.从redis中读取offset
    val topic = "ODS_BASE_DB_1118"
    val groupId = "ODS_BASE_DB_GROUP"

    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(topic,groupId)


    //3.从Kafka中消费数据
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsets != null && offsets.nonEmpty){
      kafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,topic,groupId,offsets)
    }else{
      kafkaDStream = MyKafkaUtils.getKafkaDstream(ssc,topic,groupId)
    }

    //4.提取偏移量
    var offsetRanges: Array[OffsetRange] = null
    val offsetDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )


    //5.处理数据
    //5.1转换数据结构
    val jsonObjDStream: DStream[JSONObject] = offsetDStream.map(
      consumerRecord => {
        val jsonString: String = consumerRecord.value()
        val jsonObj: JSONObject = JSON.parseObject(jsonString)
        jsonObj
      }
    )
    //5.2分流
    //jsonObjDStream.print(100) 测试消费到数据

    //定义事实表和维度表的清单
    //val dimTables : Set[String] = Set[String]("user_info","base_province","sku_info")
    //val factTables : Set[String] = Set[String]("order_info","order_detail")


    jsonObjDStream.foreachRDD(
      rdd => {
        //动态配置表清单，将表清单维护到redis中，redis shell写入，用代码再从redis中读取
        val jedis: Jedis = MyRedisUtils.getJedis()
        //读取维度表清单
        val dimKey : String = "DIM:TABLES"
        val dimTables: util.Set[String] = jedis.smembers(dimKey)
        println("dimTables : " + dimTables)
        //搞成广播变量,driver发送时只需要给每个excutor发送即可，不需要给每个task发送
        val dimTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTables)
        //读取事实表清单
        val factKey : String = "FACT:TABLES"
        val factTables: util.Set[String] = jedis.smembers(factKey)
        println("factTables: " + factTables)
        val factTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTables)
        //搞成广播变量

        MyRedisUtils.close(jedis)

        //实时处理一个分区数据不大，但离线处理一个分区数据相对较大，慎用foreachPartiton
        rdd.foreachPartition(
          itor => {
            //excutor段 每批次没分区获取一次redis连接,连接对象不能从driver端传递到excutor
            val jedis: Jedis = MyRedisUtils.getJedis()

            for (jsonObj <- itor) {
              //获取每条数据，开始进行分流工作
              //提取操作类型
              val oper: String = jsonObj.getString("type")
              //过滤数据和判断每条数据类型
              val operType: String = oper match {
                case "bootstrap-insert" => "I" //用于历史数据的引导
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null
              }

              if(operType != null){
                //获取表名
                val tableName: String = jsonObj.getString("table")

                //分流事实数据
                if(factTablesBC.value.contains(tableName)){
                  val dataObj: JSONObject = jsonObj.getJSONObject("data")
                  val id: String = jsonObj.getString("id")
                  val dwd_topic_name : String = s"DWD_${tableName.toUpperCase()}_${operType}_1118"
                  //保证分流后的数据发送到kafka中是有序的，按照key进行发送
                  //MyKafkaUtils.send(dwd_topic_name,id,dataObj.toJSONString)
                  MyKafkaUtils.send(dwd_topic_name,dataObj.toJSONString)

                }

                //分流维度数据
                //类型 :string(一条数据一个kv)  set zset list  hash
                //key : DIM:表明:id
                //value :整条数据的json
                //写入Api : set
                //读取Api :get
                //是否过期 : no
                if(dimTablesBC.value.contains(tableName)){
                  val dataObj: JSONObject = jsonObj.getJSONObject("data")
                  val id: String = dataObj.getString("id")
                  val key :String = s"DIM:${tableName.toUpperCase}:$id"

                  //val jedis: Jedis = MyRedisUtils.getJedis()
                  jedis.set(key,dataObj.toJSONString)

                  //MyRedisUtils.close(jedis)

                }

              }


            }
            //关闭redis连接
            MyRedisUtils.close(jedis)
            //刷新缓冲区
            MyKafkaUtils.flush()
          }
        )
        //提交offset
        MyOffsetUtils.saveOffset(topic,groupId,offsetRanges)
      }
    )





    ssc.start()
    ssc.awaitTermination()
  }

}
